package com.medsoft.drpcp;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.medsoft.drpcp.meta.DrpcpMetaConstant;

public class DrpcpHeart implements Runnable {
	static final Logger logger = LoggerFactory.getLogger(DrpcpHeart.class);
	@Override
	public void run() {
		Thread.currentThread().setName(DrpcpMetaConstant.HEARTBEAT_THREAD_NAME);
		while(true) {
			try {
				Thread.sleep(DrpcpMetaConstant.HEARTBEAT_INTERVAL);
				if(GlobalConfig.getServers().size() > 0) {
					logger.info("drpcp heartbeat detect...");
					for(Server server : GlobalConfig.getServers().values()) {
						for(Member member : server.getMembers().values()) {
							BlockingQueue<String> retQueue = new LinkedBlockingQueue<String>(1);
							DrpcpRouter.drpcpMultiplex.put(DrpcpMetaConstant.HEARTBEAT_MSG, retQueue);
							
							ChannelFuture cf = DrpcpRouter.callDrpcpService(member.getChannel(),DrpcpMetaConstant.HEARTBEAT_MSG);
							String retJsonStr = null;
							if (cf != null) {
								retJsonStr = retQueue.poll(GlobalConfig.timeout, TimeUnit.SECONDS);
							}
							DrpcpRouter.drpcpMultiplex.remove(DrpcpMetaConstant.HEARTBEAT_MSG);
							if(retJsonStr == null) {
								member.setConnected(false);
								logger.error("disconnected from server [" + member.getMemberName() + "],retry now...");
								Channel channel = Bootstrapper.getChannel(member.getAddress(),member.getPort());
								if(channel != null) {
									member.setChannel(channel);
									member.setConnected(true);
								} else {
									logger.error("cannot connect to server [" + member.getMemberName() + "], retry after " + DrpcpMetaConstant.HEARTBEAT_INTERVAL/1000 + " seconds.");
								}
							} else {
								logger.info("connected with server [" + member.getMemberName() + "] successfully.");
							}
						}
					}
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}
